//! 
//! Tcp Server 客户端
//! 主要与服务端进行通信和数据传输
//! 
//! tcp_server.rs
//! 
//! (RUST) alone
//! 
//! Create time: 2025-09-23		alone
//! Update time: 2025-09-29		alone
//! 

use std::{net::SocketAddr, sync::{atomic::{AtomicBool, Ordering}, Arc}, time::Duration};

use anyhow::{anyhow, Ok};
use tokio::{net::{TcpListener, TcpStream}, sync::{broadcast, watch}, time};
use tracing::{info, error};

use crate::{common::pubsub::Subscribe, net::{tcp_client::TcpClient, NetEvent}};

#[derive(Debug, Clone)]
pub struct TcpServer
{
	/// 绑定端口
	bind: SocketAddr,
	/// 结束监听
	shutdown_tx: watch::Sender<bool>,
	/// 广播订阅事件
	tx_evt: broadcast::Sender<NetEvent>,
	/// 是否已连接
	is_listened: Arc<AtomicBool>,
}

impl TcpServer
{
	pub fn new(addr: SocketAddr)
		-> Self
	{
		let (tx, _) = tokio::sync::broadcast::channel(1000);
		let (shutdown_tx, _) = watch::channel(false);
		Self 
		{ 
			bind: addr, 
			shutdown_tx,
			tx_evt: tx,
			is_listened: Arc::new(AtomicBool::new(false)), 
		}
	}

	pub async fn start(&self) 
		-> anyhow::Result<()>
	{
		if self.is_listened()
		{
			return Ok(())
		}

		let listener = match TcpListener::bind(self.bind).await
		{
			Result::Ok(result) =>
			{
				result
			}
			Result::Err(e) =>
			{
				return Err(anyhow!("绑定地址[{}]失败，错误: {}", self.bind, e));
			}
		};

		info!("Tcp 服务器监听地址: [{}]", self.bind);
		let mut shutdown_rx = self.shutdown_tx.subscribe();

		self.mark_listened();
		loop 
		{
			tokio::select! 
			{
				_ = shutdown_rx.changed() => 
				{
					info!("停止监听地址 -> {}", self.bind);
					let _ = self.tx_evt.send(NetEvent::StopListened);
					break;
				}
				res = listener.accept() => 
				{
					match res 
					{
						Result::Ok((stream, peer)) => 
						{
							let _ = self.hanlde_conn(stream, peer).await;
						}
						Err(e) => 
						{
							error!("accept error: {e}");
							time::sleep(Duration::from_millis(20)).await;
						}
					}
				}
			}
        }

		self.mark_dislistened();
		Ok(())
	}

	/// 处理连接
	async fn hanlde_conn(&self, stream: TcpStream, peer: SocketAddr) 
		-> anyhow::Result<()>
	{
		let (receiver, writer) = stream.into_split();

		let mut client = TcpClient::new(peer.to_string());
		client.set_reader(receiver).await;
		client.set_writer(writer).await;
		client.set_event(self.tx_evt.clone()).await;
		client.mark_connected();

		// 开启读任务
		let client_clone = client.clone();
		tokio::spawn(async move {
			client_clone.start();
		});

		let _ = self.tx_evt.send(NetEvent::Accept(client));

		Ok(())
	}

	fn is_listened(&self) 
		-> bool 
	{
		self.is_listened.load(std::sync::atomic::Ordering::Relaxed)
	}

	/// 设置已监听
	fn mark_listened(&self)
	{
		self.is_listened.store(true, Ordering::Release);
	}

	/// 设置未监听
	fn mark_dislistened(&self)
	{
		self.is_listened.store(false, Ordering::Release);
	}

	pub fn event(&self)
		-> broadcast::Sender<NetEvent>
	{
		self.tx_evt.clone()
	}
}

impl Subscribe for TcpServer
{
	type Event = NetEvent;

	fn subscribe(&self) -> 
		broadcast::Receiver<Self::Event> 
	{
		self.tx_evt.subscribe()	
	}
}
